#include "compi/context.h"
#include "compi/result.h"
#include "compi/tag.h"
// 使用拆分后的 P2P 接口与引擎头文件
#include "compi/p2p.h"
#include "compi/p2p_engine.h"

#include <algorithm>
#include <cassert>
#include <cstring>
#include <functional>
#include <memory>
#include <mpi.h>
#include <optional>
#include <string>
#include <vector>

namespace compi
{

namespace detail
{
// 统一 MPI 调用返回码处理：
// - 返回 true 表示 rc == MPI_SUCCESS；
// - 在 ThrowException 策略下抛异常；
// - 在 LogAndContinue 下记录并返回 false；
// - 在 AbortProcess 下调用 MPI_Abort 并终止进程。
bool
call_mpi(int rc, const char * context, MPI_Comm comm /*= MPI_COMM_WORLD*/)
{
    // 统一委托给新的接口（由 Context 注入策略）；
    // 若未指定 comm 或为 MPI_COMM_NULL，则兼容地使用 MPI_COMM_WORLD。
    MPI_Comm use_comm = (comm == MPI_COMM_NULL) ? MPI_COMM_WORLD : comm;
    return compi::call_mpi(rc, context, use_comm);
}

// 不抛异常的MPI调用包装函数，返回Result<bool>
// 参数：rc MPI返回码；context 错误上下文信息；comm MPI通信器
// 返回值：Result<bool>，成功时包含true，失败时包含错误码和错误信息
// 异常：无
Result<bool>
mpi_call_result(int rc, const char * context, MPI_Comm comm = MPI_COMM_WORLD) noexcept
{
    if (rc == MPI_SUCCESS)
        return Result<bool>::success(true);

    std::string error_msg = std::string(context) + " (rc=" + std::to_string(rc) + ")";
    return Result<bool>::error(ResultCode::MpiError, error_msg);
}
} // namespace detail

// 注意：Communicator 已别名为 Context（见 include/compi/context.h），
// 旧版 Context 的实现已移除。

// 本地辅助：查询 TAG 上限并计算命名空间步长，用于 ANY_TAG 过滤扫描
namespace
{
static int
query_tag_ub_local(MPI_Comm comm)
{
    int flag = 0;
    int * val = nullptr;
    int rc = MPI_Comm_get_attr(comm, MPI_TAG_UB, &val, &flag);
    if (rc == MPI_SUCCESS && flag && val)
        return *val;
    return 32767; // MPI 至少保证该上限
}

static int
compute_stride_local(MPI_Comm comm)
{
    const int tag_ub = query_tag_ub_local(comm);
    int candidate = std::max(64, std::min(1024, tag_ub / 8));
    if (candidate <= 0)
        candidate = 64;
    return candidate;
}
} // anonymous namespace

class TransportRequestImpl
{
public:
    MPI_Request req = MPI_REQUEST_NULL;
    bool completed = false;
    bool cancelled = false;
    // 非固定类型的异步传输需要持有缓冲区以保证生命周期
    std::shared_ptr<std::vector<std::byte>> buffer;
    // 仅用于 irecv：在完成后将 buffer 解包写入 payload
    using unpack_fn_t = void (*)(const std::vector<std::byte> &, void *);
    void * payload_ptr = nullptr;
    unpack_fn_t unpack_fn = nullptr;
    // 非固定类型 irecv 的延迟匹配状态
    bool needs_match = false;                   // 是否尚未发布接收，需要匹配探测
    int pending_src = MPI_ANY_SOURCE;           // 待匹配的源进程
    int pending_tag = MPI_ANY_TAG;              // 待匹配的消息标签
    MPI_Comm pending_comm = MPI_COMM_NULL;      // 待匹配的通信器
    MPI_Message pending_msg = MPI_MESSAGE_NULL; // 可选：已探测到的消息句柄
    int expected_bytes = 0;                     // 探测到的预期字节数（MPI_Get_count）

    // 标签管理（非阻塞时保持引用并在完成后释放）
    bool has_tag = false;
    std::optional<MessageTag> tag;
    MessageTagAllocator * alloc = nullptr;
    bool use_ns = false;                // 若为 true 则通过命名空间释放
    std::optional<TagNamespace> tag_ns; // 命名空间句柄（值语义，工厂中设置）

    // ANY_TAG 过滤（命名空间）
    bool ns_filter = false; // 若为 true，则在匹配阶段按命名空间过滤
    std::optional<TagNamespace> ns_filter_ns; // 过滤用的命名空间句柄

    // 固定类型 irecv 需要延迟发布接收，存储必要信息
    bool is_fixed = false;
    MPI_Datatype fixed_dt = MPI_BYTE;

    // 完成策略与回调链
    unsigned strategy_flags = StrategyUnpack | StrategyRelease;
    std::function<void(void *)> process_fn{};
    std::function<void(void *, const RequestStatus &)> processor_ctx{};
    std::vector<std::function<void()>> callbacks;

    // 最近一次状态
    int status_source = -1;
    int status_tag = -1;
    int status_count_bytes = 0;
    std::string error_msg;
};

static void
finalize_request(TransportRequestImpl & impl, const MPI_Status * opt_status, bool allow_throw)
{
    impl.completed = true;
    if (opt_status)
    {
        impl.status_source = opt_status->MPI_SOURCE;
        impl.status_tag = opt_status->MPI_TAG;
        // 计算字节数
        int count_elems = 0;
        if (impl.is_fixed && impl.fixed_dt != MPI_BYTE)
        {
            MPI_Get_count(const_cast<MPI_Status *>(opt_status), impl.fixed_dt, &count_elems);
            int elsize = 0;
            MPI_Type_size(impl.fixed_dt, &elsize);
            impl.status_count_bytes = count_elems * elsize;
        }
        else
        {
            impl.status_count_bytes = impl.expected_bytes;
        }
    }
    else
    {
        // 无状态时尽力填充字节数（如取消或未探测路径）
        if (impl.is_fixed && impl.fixed_dt != MPI_BYTE)
        {
            int elsize = 0;
            MPI_Type_size(impl.fixed_dt, &elsize);
            impl.status_count_bytes = elsize; // 单元素固定类型
        }
        else
        {
            // 对非固定类型，已知 expected_bytes 则使用；否则保持为 0
            // impl.status_count_bytes = impl.expected_bytes; // 可能为 0
        }
    }
    bool had_exception = false;
    // 解包
    if ((impl.strategy_flags & StrategyUnpack) && impl.unpack_fn && impl.payload_ptr && impl.buffer)
    {
        try
        {
            impl.unpack_fn(*impl.buffer, impl.payload_ptr);
        }
        catch (const std::exception & e)
        {
            impl.error_msg = e.what();
            had_exception = true;
            if (allow_throw)
                throw;
        }
        catch (...)
        {
            impl.error_msg = "Unknown exception in Unpack";
            had_exception = true;
            if (allow_throw)
                throw;
        }
    }
    // 处理
    if (!had_exception && (impl.strategy_flags & StrategyProcess))
    {
        if (impl.process_fn)
        {
            try
            {
                impl.process_fn(impl.payload_ptr);
            }
            catch (const std::exception & e)
            {
                impl.error_msg = e.what();
                had_exception = true;
                if (allow_throw)
                    throw;
            }
            catch (...)
            {
                impl.error_msg = "Unknown exception in Process";
                had_exception = true;
                if (allow_throw)
                    throw;
            }
        }
    }
    // 带上下文的处理器：无论策略是否包含 Process，只要设置了就调用
    if (!had_exception && impl.processor_ctx)
    {
        RequestStatus st{};
        st.source = impl.status_source;
        st.tag = impl.status_tag;
        st.count_bytes = impl.status_count_bytes;
        st.cancelled = impl.cancelled;
        if (!impl.error_msg.empty())
            st.error = impl.error_msg;
        try
        {
            impl.processor_ctx(impl.payload_ptr, st);
        }
        catch (const std::exception & e)
        {
            impl.error_msg = e.what();
            had_exception = true;
            if (allow_throw)
                throw;
        }
        catch (...)
        {
            impl.error_msg = "Unknown exception in Processor";
            had_exception = true;
            if (allow_throw)
                throw;
        }
    }
    // 释放资源
    if (!had_exception && (impl.strategy_flags & StrategyRelease))
    {
        if (impl.buffer)
            impl.buffer.reset();
        impl.payload_ptr = nullptr;
        impl.unpack_fn = nullptr;
        if (impl.has_tag && impl.tag.has_value())
        {
            if (impl.use_ns && impl.tag_ns)
                impl.tag_ns->release(*impl.tag);
            else if (impl.alloc)
                impl.alloc->release(*impl.tag);
            impl.has_tag = false;
            impl.tag.reset();
        }
    }
    // 回调链
    if (!had_exception)
    {
        for (auto & cb : impl.callbacks)
        {
            if (cb)
                cb();
        }
    }
}

// 等待请求完成；对于非固定类型的 irecv，在完成后执行解包写入用户提供的 payload。
// 参数：无
// 返回值：true 表示成功完成；失败时根据策略可能抛出或终止。
// 异常：在策略为抛异常时可能抛出。

// 非抛出等待：阻塞直到完成并返回统一的 Result<void>。
Result<void>
TransportRequest::wait() noexcept
{
    if (!_impl)
    {
        return Result<void>::error(ResultCode::InvalidRequest,
                                   "TransportRequest has no implementation",
                                   "TransportRequest::wait");
    }

    auto & impl = *_impl;
    if (impl.completed)
    {
        if (impl.cancelled)
            return Result<void>::error(
                ResultCode::Cancelled, "Request was cancelled", "TransportRequest::wait");
        if (!impl.error_msg.empty())
            return Result<void>::error(
                ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
        return Result<void>::success("TransportRequest::wait");
    }

    // 若尚未发布接收且需要匹配，则阻塞匹配并接收
    if (impl.req == MPI_REQUEST_NULL && impl.needs_match)
    {
        MPI_Status status{};
        int probed = 0;
        while (true)
        {
            const int iprobe_rc =
                MPI_Iprobe(impl.pending_src, MPI_ANY_TAG, impl.pending_comm, &probed, &status);
            if (iprobe_rc != MPI_SUCCESS)
            {
                impl.error_msg = "MPI_Iprobe failed";
                return Result<void>::error(
                    ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
            }
            if (probed &&
                (!impl.ns_filter || (impl.ns_filter_ns && impl.ns_filter_ns->owns(status.MPI_TAG))))
                break;
            if (impl.ns_filter)
            {
                const int stride = compute_stride_local(impl.pending_comm);
                const int ns_id = static_cast<int>(impl.ns_filter_ns->id());
                MPI_Status st2{};
                for (int k = 1; k < stride; ++k)
                {
                    int p2 = 0;
                    const int tag = ns_id * stride + k;
                    const int rc2 = MPI_Iprobe(impl.pending_src, tag, impl.pending_comm, &p2, &st2);
                    if (rc2 != MPI_SUCCESS)
                    {
                        impl.error_msg = "MPI_Iprobe failed";
                        return Result<void>::error(
                            ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
                    }
                    if (p2)
                    {
                        status = st2;
                        probed = 1;
                        break;
                    }
                }
                if (probed)
                    break;
            }
        }
        // 固定类型：阻塞接收
        if (impl.is_fixed)
        {
            const int rc = MPI_Recv(impl.payload_ptr,
                                    1,
                                    impl.fixed_dt,
                                    impl.pending_src,
                                    status.MPI_TAG,
                                    impl.pending_comm,
                                    &status);
            if (rc != MPI_SUCCESS)
            {
                impl.error_msg = "MPI_Recv failed (固定类型 ANY_TAG 过滤)";
                return Result<void>::error(
                    ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
            }
            impl.needs_match = false;
            finalize_request(impl, &status, false);
            if (impl.cancelled)
                return Result<void>::error(
                    ResultCode::Cancelled, "Request was cancelled", "TransportRequest::wait");
            if (!impl.error_msg.empty())
                return Result<void>::error(
                    ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
            return Result<void>::success("TransportRequest::wait");
        }
        // 非固定类型：Mprobe/Mrecv
        MPI_Message msg = MPI_MESSAGE_NULL;
        const int mprobe_rc =
            MPI_Mprobe(impl.pending_src, status.MPI_TAG, impl.pending_comm, &msg, &status);
        if (mprobe_rc != MPI_SUCCESS)
        {
            impl.error_msg = "MPI_Mprobe failed";
            return Result<void>::error(
                ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
        }
        int count_bytes = 0;
        const int count_rc = MPI_Get_count(&status, MPI_BYTE, &count_bytes);
        if (count_rc != MPI_SUCCESS)
        {
            impl.error_msg = "MPI_Get_count failed";
            return Result<void>::error(
                ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
        }
        impl.buffer =
            std::make_shared<std::vector<std::byte>>(static_cast<std::size_t>(count_bytes));
        const int mrecv_rc = MPI_Mrecv(impl.buffer->data(), count_bytes, MPI_BYTE, &msg, &status);
        if (mrecv_rc != MPI_SUCCESS)
        {
            impl.error_msg = "MPI_Mrecv failed";
            return Result<void>::error(
                ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
        }
        impl.needs_match = false;
        impl.expected_bytes = count_bytes;
        finalize_request(impl, &status, false);
        if (impl.cancelled)
            return Result<void>::error(
                ResultCode::Cancelled, "Request was cancelled", "TransportRequest::wait");
        if (!impl.error_msg.empty())
            return Result<void>::error(
                ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
        return Result<void>::success("TransportRequest::wait");
    }

    // 已发布请求：等待完成
    MPI_Status status{};
    const int rc = MPI_Wait(&impl.req, &status);
    if (rc != MPI_SUCCESS)
    {
        impl.error_msg = "MPI_Wait failed";
        return Result<void>::error(ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
    }
    finalize_request(impl, &status, false);
    if (impl.cancelled)
        return Result<void>::error(
            ResultCode::Cancelled, "Request was cancelled", "TransportRequest::wait");
    if (!impl.error_msg.empty())
        return Result<void>::error(ResultCode::MpiError, impl.error_msg, "TransportRequest::wait");
    return Result<void>::success("TransportRequest::wait");
}

// 非阻塞查询请求是否完成。
// 参数：无
// 返回值：true 表示请求已完成；false 表示未完成或出现测试错误。
// 异常：无
bool
TransportRequest::completed() noexcept
{
    if (!_impl)
        return false;
    auto & impl = *_impl;
    if (impl.completed)
        return true;

    // 若尚未发布接收且需要匹配，则尝试非阻塞匹配并发布接收
    if (impl.req == MPI_REQUEST_NULL && impl.needs_match)
    {
        MPI_Status status{};
        int probed = 0;
        // 先用 Iprobe 在 ANY_TAG 下检查是否存在属于命名空间的消息
        const int iprobe_rc =
            MPI_Iprobe(impl.pending_src, MPI_ANY_TAG, impl.pending_comm, &probed, &status);
        if (!call_mpi(iprobe_rc, "MPI_Iprobe failed (completed: match phase)", impl.pending_comm))
        {
            impl.error_msg = "MPI_Iprobe failed (completed: match phase)";
            return false;
        }
        if (probed &&
            (!impl.ns_filter || (impl.ns_filter_ns && impl.ns_filter_ns->owns(status.MPI_TAG))))
        {
            // 若为固定类型，直接发布 Irecv
            if (impl.is_fixed)
            {
                const int rc = MPI_Irecv(impl.payload_ptr,
                                         1,
                                         impl.fixed_dt,
                                         impl.pending_src,
                                         status.MPI_TAG,
                                         impl.pending_comm,
                                         &impl.req);
                if (!call_mpi(rc,
                              "MPI_Irecv failed (completed: fixed ANY_TAG publish)",
                              impl.pending_comm))
                {
                    impl.error_msg = "MPI_Irecv failed (completed: fixed ANY_TAG publish)";
                    return false;
                }
                impl.needs_match = false;
            }
            else
            {
                // 非固定类型：以具体标签执行 Improbe/Imrecv
                MPI_Message msg = MPI_MESSAGE_NULL;
                int ready = 0;
                const int improbe_rc = MPI_Improbe(
                    impl.pending_src, status.MPI_TAG, impl.pending_comm, &ready, &msg, &status);
                if (!call_mpi(improbe_rc,
                              "MPI_Improbe failed (completed: serialized ANY_TAG publish)",
                              impl.pending_comm) ||
                    !ready)
                {
                    impl.error_msg =
                        "MPI_Improbe failed or not ready (completed: serialized ANY_TAG publish)";
                    return false;
                }
                int count_bytes = 0;
                const int count_rc = MPI_Get_count(&status, MPI_BYTE, &count_bytes);
                if (!call_mpi(count_rc,
                              "MPI_Get_count failed (completed: serialized ANY_TAG publish)",
                              impl.pending_comm))
                {
                    impl.error_msg = "MPI_Get_count failed (completed: serialized ANY_TAG publish)";
                    return false;
                }
                impl.buffer =
                    std::make_shared<std::vector<std::byte>>(static_cast<std::size_t>(count_bytes));
                const int imrecv_rc =
                    MPI_Imrecv(impl.buffer->data(), count_bytes, MPI_BYTE, &msg, &impl.req);
                if (!call_mpi(imrecv_rc,
                              "MPI_Imrecv failed (completed: serialized ANY_TAG publish)",
                              impl.pending_comm))
                {
                    impl.error_msg = "MPI_Imrecv failed (completed: serialized ANY_TAG publish)";
                    return false;
                }
                impl.needs_match = false;
                impl.expected_bytes = count_bytes;
            }
        }
        else if (impl.ns_filter)
        {
            // 使用命名空间的范围进行安全探测，避免越界或非属标签
            MPI_Status st2{};
            MessageTag out_tag(0, impl.ns_filter_ns->id());
            const bool found =
                impl.ns_filter_ns->iprobe(impl.pending_comm, impl.pending_src, out_tag);
            if (found)
            {
                const int tag_val = static_cast<int>(out_tag);
                if (impl.is_fixed)
                {
                    const int irc = MPI_Irecv(impl.payload_ptr,
                                              1,
                                              impl.fixed_dt,
                                              impl.pending_src,
                                              tag_val,
                                              impl.pending_comm,
                                              &impl.req);
                    if (!call_mpi(irc,
                                  "MPI_Irecv failed (completed: fixed ANY_TAG namespace publish)",
                                  impl.pending_comm))
                    {
                        impl.error_msg =
                            "MPI_Irecv failed (completed: fixed ANY_TAG namespace publish)";
                        return false;
                    }
                    impl.needs_match = false;
                }
                else
                {
                    MPI_Message msg = MPI_MESSAGE_NULL;
                    int ready = 0;
                    const int improbe_rc = MPI_Improbe(
                        impl.pending_src, tag_val, impl.pending_comm, &ready, &msg, &st2);
                    if (!call_mpi(
                            improbe_rc,
                            "MPI_Improbe failed (completed: serialized ANY_TAG namespace publish)",
                            impl.pending_comm) ||
                        !ready)
                    {
                        impl.error_msg = "MPI_Improbe failed or not ready (completed: "
                                         "serialized ANY_TAG namespace publish)";
                        return false;
                    }
                    int count_bytes = 0;
                    const int count_rc = MPI_Get_count(&st2, MPI_BYTE, &count_bytes);
                    if (!call_mpi(count_rc,
                                  "MPI_Get_count failed (completed: serialized ANY_TAG namespace "
                                  "publish)",
                                  impl.pending_comm))
                    {
                        impl.error_msg = "MPI_Get_count failed (completed: serialized ANY_TAG "
                                         "namespace publish)";
                        return false;
                    }
                    impl.buffer = std::make_shared<std::vector<std::byte>>(
                        static_cast<std::size_t>(count_bytes));
                    const int imrecv_rc =
                        MPI_Imrecv(impl.buffer->data(), count_bytes, MPI_BYTE, &msg, &impl.req);
                    if (!call_mpi(
                            imrecv_rc,
                            "MPI_Imrecv failed (completed: serialized ANY_TAG namespace publish)",
                            impl.pending_comm))
                    {
                        impl.error_msg =
                            "MPI_Imrecv failed (completed: serialized ANY_TAG namespace publish)";
                        return false;
                    }
                    impl.needs_match = false;
                    impl.expected_bytes = count_bytes;
                }
            }
        }
    }

    if (impl.req == MPI_REQUEST_NULL)
    {
        // 尚未发布任何具体请求，不能视为完成
        return false;
    }
    int flag = 0;
    MPI_Status status{};
    const int rc = MPI_Test(&impl.req, &flag, &status);
    if (!call_mpi(rc, "MPI_Test failed (completed: test phase)", impl.pending_comm))
    {
        impl.error_msg = "MPI_Test failed (completed: test phase)";
        return false;
    }
    if (flag)
    {
        finalize_request(impl, &status, true);
        return true;
    }
    return false;
}

bool
TransportRequest::cancel() noexcept
{
    if (!_impl)
        return false;
    auto & impl = *_impl;
    if (impl.completed || impl.cancelled)
        return false;
    // 若尚未发布接收，仅标记取消并释放标签
    if (impl.req == MPI_REQUEST_NULL && impl.needs_match)
    {
        impl.cancelled = true;
        impl.status_source = -1;
        impl.status_tag = impl.pending_tag;
        impl.status_count_bytes = 0;
        if (impl.has_tag && impl.tag.has_value())
        {
            if (impl.use_ns && impl.tag_ns)
                impl.tag_ns->release(*impl.tag);
            else if (impl.alloc)
                impl.alloc->release(*impl.tag);
            impl.has_tag = false;
            impl.tag.reset();
        }
        impl.buffer.reset();
        impl.payload_ptr = nullptr;
        impl.unpack_fn = nullptr;
        finalize_request(impl, nullptr, false);
        return true;
    }
    // 尝试取消已发布的请求（最佳努力）
    if (impl.req != MPI_REQUEST_NULL)
    {
        const int rc = MPI_Cancel(&impl.req);
        if (!call_mpi(rc, "MPI_Cancel failed", impl.pending_comm))
        {
            impl.error_msg = "MPI_Cancel failed";
            return false;
        }
        int flag = 0;
        MPI_Status st{};
        const int trc = MPI_Test(&impl.req, &flag, &st); // 推进取消完成
        if (!call_mpi(trc, "MPI_Test failed (cancel)", impl.pending_comm))
        {
            impl.error_msg = "MPI_Test failed (cancel)";
            return false;
        }
        impl.cancelled = true;
        finalize_request(impl, nullptr, false);
        return true;
    }
    return false;
}

bool
TransportRequest::poll() noexcept
{
    if (!_impl)
        return false;
    auto & impl = *_impl;
    if (impl.completed || impl.cancelled)
        return false;
    bool progressed = false;
    // 先尝试非阻塞匹配并发布接收
    if (impl.req == MPI_REQUEST_NULL && impl.needs_match)
    {
        MPI_Status status{};
        int probed = 0;
        const int iprobe_rc =
            MPI_Iprobe(impl.pending_src, MPI_ANY_TAG, impl.pending_comm, &probed, &status);
        if (call_mpi(iprobe_rc, "MPI_Iprobe failed (poll: match phase)", impl.pending_comm) &&
            probed &&
            (!impl.ns_filter || (impl.ns_filter_ns && impl.ns_filter_ns->owns(status.MPI_TAG))))
        {
            if (impl.is_fixed)
            {
                const int rc = MPI_Irecv(impl.payload_ptr,
                                         1,
                                         impl.fixed_dt,
                                         impl.pending_src,
                                         status.MPI_TAG,
                                         impl.pending_comm,
                                         &impl.req);
                if (call_mpi(
                        rc, "MPI_Irecv failed (poll: fixed ANY_TAG publish)", impl.pending_comm))
                {
                    impl.needs_match = false;
                    progressed = true;
                }
                else
                {
                    impl.error_msg = "MPI_Irecv failed (poll: fixed ANY_TAG publish)";
                }
            }
            else
            {
                MPI_Message msg = MPI_MESSAGE_NULL;
                int ready = 0;
                const int improbe_rc = MPI_Improbe(
                    impl.pending_src, status.MPI_TAG, impl.pending_comm, &ready, &msg, &status);
                if (call_mpi(improbe_rc,
                             "MPI_Improbe failed (poll: serialized ANY_TAG publish)",
                             impl.pending_comm) &&
                    ready)
                {
                    int count_bytes = 0;
                    const int count_rc = MPI_Get_count(&status, MPI_BYTE, &count_bytes);
                    if (call_mpi(count_rc,
                                 "MPI_Get_count failed (poll: serialized ANY_TAG publish)",
                                 impl.pending_comm))
                    {
                        impl.buffer = std::make_shared<std::vector<std::byte>>(
                            static_cast<std::size_t>(count_bytes));
                        const int imrecv_rc =
                            MPI_Imrecv(impl.buffer->data(), count_bytes, MPI_BYTE, &msg, &impl.req);
                        if (call_mpi(imrecv_rc,
                                     "MPI_Imrecv failed (poll: serialized ANY_TAG publish)",
                                     impl.pending_comm))
                        {
                            impl.needs_match = false;
                            impl.expected_bytes = count_bytes;
                            progressed = true;
                        }
                        else
                        {
                            impl.error_msg = "MPI_Imrecv failed (poll: serialized ANY_TAG publish)";
                        }
                    }
                    else
                    {
                        impl.error_msg = "MPI_Get_count failed (poll: serialized ANY_TAG publish)";
                    }
                }
                else if (!call_mpi(improbe_rc,
                                   "MPI_Improbe failed (poll: serialized ANY_TAG publish)",
                                   impl.pending_comm))
                {
                    impl.error_msg = "MPI_Improbe failed (poll: serialized ANY_TAG publish)";
                }
            }
        }
        else if (call_mpi(iprobe_rc, "MPI_Iprobe failed (poll: match phase)", impl.pending_comm) &&
                 impl.ns_filter)
        {
            // 规避队首阻塞：扫描命名空间可能的标签，若存在则直接发布接收
            const int stride = compute_stride_local(impl.pending_comm);
            const int ns_id = static_cast<int>(impl.ns_filter_ns->id());
            MPI_Status st2{};
            for (int k = 1; k < stride; ++k)
            {
                const int tag = ns_id * stride + k;
                int p2 = 0;
                const int rc2 = MPI_Iprobe(impl.pending_src, tag, impl.pending_comm, &p2, &st2);
                if (call_mpi(rc2, "MPI_Iprobe failed (poll: scan tags)", impl.pending_comm) && p2)
                {
                    if (impl.is_fixed)
                    {
                        const int irc = MPI_Irecv(impl.payload_ptr,
                                                  1,
                                                  impl.fixed_dt,
                                                  impl.pending_src,
                                                  tag,
                                                  impl.pending_comm,
                                                  &impl.req);
                        if (call_mpi(irc,
                                     "MPI_Irecv failed (poll: fixed ANY_TAG scan publish)",
                                     impl.pending_comm))
                        {
                            impl.needs_match = false;
                            progressed = true;
                        }
                        else
                        {
                            impl.error_msg = "MPI_Irecv failed (poll: fixed ANY_TAG scan publish)";
                        }
                        break;
                    }
                    else
                    {
                        MPI_Message msg = MPI_MESSAGE_NULL;
                        int ready = 0;
                        const int improbe_rc = MPI_Improbe(
                            impl.pending_src, tag, impl.pending_comm, &ready, &msg, &st2);
                        if (call_mpi(improbe_rc,
                                     "MPI_Improbe failed (poll: serialized ANY_TAG scan publish)",
                                     impl.pending_comm) &&
                            ready)
                        {
                            int count_bytes = 0;
                            const int count_rc = MPI_Get_count(&st2, MPI_BYTE, &count_bytes);
                            if (call_mpi(
                                    count_rc,
                                    "MPI_Get_count failed (poll: serialized ANY_TAG scan publish)",
                                    impl.pending_comm))
                            {
                                impl.buffer = std::make_shared<std::vector<std::byte>>(
                                    static_cast<std::size_t>(count_bytes));
                                const int imrecv_rc = MPI_Imrecv(
                                    impl.buffer->data(), count_bytes, MPI_BYTE, &msg, &impl.req);
                                if (call_mpi(
                                        imrecv_rc,
                                        "MPI_Imrecv failed (poll: serialized ANY_TAG scan publish)",
                                        impl.pending_comm))
                                {
                                    impl.needs_match = false;
                                    impl.expected_bytes = count_bytes;
                                    progressed = true;
                                }
                                else
                                {
                                    impl.error_msg =
                                        "MPI_Imrecv failed (poll: serialized ANY_TAG scan publish)";
                                }
                            }
                            else
                            {
                                impl.error_msg =
                                    "MPI_Get_count failed (poll: serialized ANY_TAG scan publish)";
                            }
                            break;
                        }
                        else if (!call_mpi(
                                     improbe_rc,
                                     "MPI_Improbe failed (poll: serialized ANY_TAG scan publish)",
                                     impl.pending_comm))
                        {
                            impl.error_msg =
                                "MPI_Improbe failed (poll: serialized ANY_TAG scan publish)";
                            break;
                        }
                    }
                }
            }
        }
        else if (!call_mpi(iprobe_rc, "MPI_Iprobe failed (poll: match phase)", impl.pending_comm))
        {
            impl.error_msg = "MPI_Iprobe failed (poll: match phase)";
        }
    }
    // 再测试完成
    int flag = 0;
    MPI_Status st{};
    if (impl.req != MPI_REQUEST_NULL)
    {
        const int trc = MPI_Test(&impl.req, &flag, &st);
        if (call_mpi(trc, "MPI_Test failed (poll: test phase)", impl.pending_comm) && flag)
        {
            finalize_request(impl, &st, false);
            progressed = true;
        }
        else if (!call_mpi(trc, "MPI_Test failed (poll: test phase)", impl.pending_comm))
        {
            impl.error_msg = "MPI_Test failed (poll: test phase)";
        }
    }
    return progressed;
}

RequestStatus
TransportRequest::status() const noexcept
{
    RequestStatus st{};
    if (!_impl)
        return st;
    st.source = _impl->status_source;
    st.tag = _impl->status_tag;
    st.count_bytes = _impl->status_count_bytes;
    st.cancelled = _impl->cancelled;
    if (!_impl->error_msg.empty())
        st.error = _impl->error_msg;
    return st;
}

Result<RequestStatus>
TransportRequest::result() const noexcept
{
    if (!_impl)
    {
        return Result<RequestStatus>::error(ResultCode::InvalidRequest,
                                            "TransportRequest has no implementation");
    }

    RequestStatus st{};
    st.source = _impl->status_source;
    st.tag = _impl->status_tag;
    st.count_bytes = _impl->status_count_bytes;
    st.cancelled = _impl->cancelled;

    if (_impl->cancelled)
    {
        return Result<RequestStatus>::error(ResultCode::Cancelled, "Request was cancelled");
    }

    if (!_impl->error_msg.empty())
    {
        st.error = _impl->error_msg;
        return Result<RequestStatus>::error(ResultCode::MpiError, _impl->error_msg);
    }

    return Result<RequestStatus>::success(std::move(st));
}

void
TransportRequest::set_processor(std::function<void(void *)> process) noexcept
{
    if (_impl)
        _impl->process_fn = std::move(process);
}

void
TransportRequest::set_processor(
    std::function<void(void *, const RequestStatus &)> processor) noexcept
{
    if (_impl)
        _impl->processor_ctx = std::move(processor);
}

void
TransportRequest::add_callback(std::function<void()> cb)
{
    if (_impl)
        _impl->callbacks.push_back(std::move(cb));
}

void
TransportRequest::set_strategy(unsigned strategy_flags) noexcept
{
    if (_impl)
        _impl->strategy_flags = strategy_flags;
}

bool
RequestGroup::wait_all()
{
    bool all_ok = true;
    for (auto & r : _requests)
    {
        Result<void> st = r.wait();
        // 在批量等待语义中，将取消视为“完成”的一种形式
        if (!(st.is_success() || st.error_code() == ResultCode::Cancelled))
            all_ok = false;
    }
    return all_ok;
}

std::vector<std::size_t>
RequestGroup::test_some()
{
    std::vector<std::size_t> done;
    done.reserve(_requests.size());
    for (std::size_t i = 0; i < _requests.size(); ++i)
    {
        if (_requests[i].completed())
            done.push_back(i);
    }
    return done;
}

bool
RequestGroup::poll_all() noexcept
{
    bool progressed = false;
    for (auto & r : _requests)
    {
        progressed = r.poll() || progressed;
    }
    return progressed;
}
} // namespace compi

// 轻量工厂实现：为固定类型提供非阻塞发送/接收请求构造
// 注意：这些实现严格遵循 TransportRequest 的内部行为与标签管理约定。
namespace compi
{
TransportRequest
irecv_fixed(Context & ctx, int src, void * payload, MPI_Datatype dt, TagNamespace ns) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    // 固定类型 ANY_TAG 过滤接收：延迟发布接收，等待匹配阶段选择具体标签
    impl->is_fixed = true;
    impl->fixed_dt = dt;
    impl->payload_ptr = payload;
    impl->needs_match = true;
    impl->pending_src = src;
    impl->pending_tag = MPI_ANY_TAG;
    impl->pending_comm = ctx.comm();
    // 命名空间过滤：仅匹配该命名空间内的标签
    impl->ns_filter = true;
    impl->ns_filter_ns = ns;
    // 初始状态元数据
    impl->status_source = -1;
    impl->status_tag = MPI_ANY_TAG;
    // 默认完成策略：Unpack（对固定类型无效）+ Release（无标签，安全）
    return TransportRequest{impl};
}

TransportRequest
isend_fixed(Context & ctx,
            int dest,
            const void * payload,
            MPI_Datatype dt,
            TagNamespace ns,
            SendMode mode) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    // 申请并保留命名空间标签，用于非阻塞发送直至完成
    MessageTag tag = ns.request();
    ns.retain(tag);
    impl->has_tag = true;
    impl->tag = tag;
    impl->use_ns = true;
    impl->tag_ns = ns;
    impl->pending_comm = ctx.comm();
    impl->is_fixed = true;
    impl->fixed_dt = dt;
    impl->status_source = -1;
    impl->status_tag = static_cast<int>(tag);

    MPI_Request req = MPI_REQUEST_NULL;
    const int rc =
        (mode == SendMode::Synchronous)
            ? MPI_Issend(
                  const_cast<void *>(payload), 1, dt, dest, static_cast<int>(tag), ctx.comm(), &req)
            : MPI_Isend(const_cast<void *>(payload),
                        1,
                        dt,
                        dest,
                        static_cast<int>(tag),
                        ctx.comm(),
                        &req);

    if (!detail::call_mpi(rc, "MPI_Isend/Issend failed (isend_fixed)", ctx.comm()))
    {
        // 记录错误并立即完成以触发资源释放（包含标签释放）
        impl->error_msg = "MPI_Isend/Issend failed (isend_fixed)";
        finalize_request(*impl, nullptr, false);
        return TransportRequest{impl};
    }

    impl->req = req;
    return TransportRequest{impl};
}

TransportRequest
isend_fixed(Context & ctx,
            int dest,
            const void * payload,
            MPI_Datatype dt,
            const TagLease & lease,
            SendMode mode) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    // 使用已有租约标签，避免重复 retain/release；完成时不自动释放标签
    const MessageTag tag = lease.tag();
    impl->has_tag = true;
    impl->tag = tag;
    impl->use_ns = false;  // 不通过命名空间释放
    impl->alloc = nullptr; // 明确不使用分配器释放
    impl->pending_comm = ctx.comm();
    impl->is_fixed = true;
    impl->fixed_dt = dt;
    impl->status_source = -1;
    impl->status_tag = static_cast<int>(tag);

    MPI_Request req = MPI_REQUEST_NULL;
    const int rc =
        (mode == SendMode::Synchronous)
            ? MPI_Issend(
                  const_cast<void *>(payload), 1, dt, dest, static_cast<int>(tag), ctx.comm(), &req)
            : MPI_Isend(const_cast<void *>(payload),
                        1,
                        dt,
                        dest,
                        static_cast<int>(tag),
                        ctx.comm(),
                        &req);

    if (!detail::call_mpi(rc, "MPI_Isend/Issend failed (isend_fixed:lease)", ctx.comm()))
    {
        impl->error_msg = "MPI_Isend/Issend failed (isend_fixed:lease)";
        finalize_request(*impl, nullptr, false);
        return TransportRequest{impl};
    }

    impl->req = req;
    return TransportRequest{impl};
}
} // namespace compi

namespace compi
{
// 序列化负载的解包函数：写入 std::string
namespace
{
static void
unpack_to_string(const std::vector<std::byte> & buf, void * payload)
{
    auto * str = static_cast<std::string *>(payload);
    str->assign(reinterpret_cast<const char *>(buf.data()), buf.size());
}

static void
unpack_to_bytes(const std::vector<std::byte> & buf, void * payload)
{
    auto * vec = static_cast<std::vector<std::byte> *>(payload);
    *vec = buf; // 拷贝字节向量
}
} // anonymous namespace

TransportRequest
irecv_serialized(Context & ctx, int src, std::string & payload, TagNamespace ns) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    // 非固定类型 ANY_TAG 过滤接收：延迟匹配，完成后按解包函数写入 payload
    impl->is_fixed = false;
    impl->payload_ptr = &payload;
    impl->unpack_fn = &unpack_to_string;
    impl->needs_match = true;
    impl->pending_src = src;
    impl->pending_tag = MPI_ANY_TAG;
    impl->pending_comm = ctx.comm();
    // 命名空间过滤
    impl->ns_filter = true;
    impl->ns_filter_ns = ns;
    // 初始状态元数据
    impl->status_source = -1;
    impl->status_tag = MPI_ANY_TAG;
    // 完成策略：默认包含解包与释放
    return TransportRequest{impl};
}

TransportRequest
irecv_serialized(Context & ctx, int src, std::vector<std::byte> & payload, TagNamespace ns) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    impl->is_fixed = false;
    impl->payload_ptr = &payload;
    impl->unpack_fn = &unpack_to_bytes;
    impl->needs_match = true;
    impl->pending_src = src;
    impl->pending_tag = MPI_ANY_TAG;
    impl->pending_comm = ctx.comm();
    impl->ns_filter = true;
    impl->ns_filter_ns = ns;
    impl->status_source = -1;
    impl->status_tag = MPI_ANY_TAG;
    return TransportRequest{impl};
}

TransportRequest
isend_serialized(
    Context & ctx, int dest, const std::string & payload, TagNamespace ns, SendMode mode) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    // 申请并保留命名空间标签
    MessageTag tag = ns.request();
    ns.retain(tag);
    impl->has_tag = true;
    impl->tag = tag;
    impl->use_ns = true;
    impl->tag_ns = ns;
    impl->pending_comm = ctx.comm();
    impl->is_fixed = false;
    impl->status_source = -1;
    impl->status_tag = static_cast<int>(tag);

    // 复制负载到内部缓冲，保证非阻塞发送期间生命周期
    impl->buffer = std::make_shared<std::vector<std::byte>>(payload.size());
    std::memcpy(impl->buffer->data(), payload.data(), payload.size());

    MPI_Request req = MPI_REQUEST_NULL;
    const int rc = (mode == SendMode::Synchronous)
                       ? MPI_Issend(impl->buffer->data(),
                                    static_cast<int>(impl->buffer->size()),
                                    MPI_BYTE,
                                    dest,
                                    static_cast<int>(tag),
                                    ctx.comm(),
                                    &req)
                       : MPI_Isend(impl->buffer->data(),
                                   static_cast<int>(impl->buffer->size()),
                                   MPI_BYTE,
                                   dest,
                                   static_cast<int>(tag),
                                   ctx.comm(),
                                   &req);

    if (!detail::call_mpi(rc, "MPI_Isend/Issend failed (isend_serialized:string)", ctx.comm()))
    {
        impl->error_msg = "MPI_Isend/Issend failed (isend_serialized:string)";
        finalize_request(*impl, nullptr, false);
        return TransportRequest{impl};
    }

    impl->req = req;
    return TransportRequest{impl};
}

TransportRequest
isend_serialized(Context & ctx,
                 int dest,
                 const std::vector<std::byte> & payload,
                 TagNamespace ns,
                 SendMode mode) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    MessageTag tag = ns.request();
    ns.retain(tag);
    impl->has_tag = true;
    impl->tag = tag;
    impl->use_ns = true;
    impl->tag_ns = ns;
    impl->pending_comm = ctx.comm();
    impl->is_fixed = false;
    impl->status_source = -1;
    impl->status_tag = static_cast<int>(tag);

    impl->buffer = std::make_shared<std::vector<std::byte>>(payload);

    MPI_Request req = MPI_REQUEST_NULL;
    const int rc = (mode == SendMode::Synchronous)
                       ? MPI_Issend(impl->buffer->data(),
                                    static_cast<int>(impl->buffer->size()),
                                    MPI_BYTE,
                                    dest,
                                    static_cast<int>(tag),
                                    ctx.comm(),
                                    &req)
                       : MPI_Isend(impl->buffer->data(),
                                   static_cast<int>(impl->buffer->size()),
                                   MPI_BYTE,
                                   dest,
                                   static_cast<int>(tag),
                                   ctx.comm(),
                                   &req);

    if (!detail::call_mpi(rc, "MPI_Isend/Issend failed (isend_serialized:bytes)", ctx.comm()))
    {
        impl->error_msg = "MPI_Isend/Issend failed (isend_serialized:bytes)";
        finalize_request(*impl, nullptr, false);
        return TransportRequest{impl};
    }

    impl->req = req;
    return TransportRequest{impl};
}

TransportRequest
isend_serialized(Context & ctx,
                 int dest,
                 const std::string & payload,
                 const TagLease & lease,
                 SendMode mode) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    const MessageTag tag = lease.tag();
    impl->has_tag = true;
    impl->tag = tag;
    impl->use_ns = false;
    impl->alloc = nullptr;
    impl->pending_comm = ctx.comm();
    impl->is_fixed = false;
    impl->status_source = -1;
    impl->status_tag = static_cast<int>(tag);

    impl->buffer = std::make_shared<std::vector<std::byte>>(payload.size());
    std::memcpy(impl->buffer->data(), payload.data(), payload.size());

    MPI_Request req = MPI_REQUEST_NULL;
    const int rc = (mode == SendMode::Synchronous)
                       ? MPI_Issend(impl->buffer->data(),
                                    static_cast<int>(impl->buffer->size()),
                                    MPI_BYTE,
                                    dest,
                                    static_cast<int>(tag),
                                    ctx.comm(),
                                    &req)
                       : MPI_Isend(impl->buffer->data(),
                                   static_cast<int>(impl->buffer->size()),
                                   MPI_BYTE,
                                   dest,
                                   static_cast<int>(tag),
                                   ctx.comm(),
                                   &req);

    if (!detail::call_mpi(
            rc, "MPI_Isend/Issend failed (isend_serialized:string:lease)", ctx.comm()))
    {
        impl->error_msg = "MPI_Isend/Issend failed (isend_serialized:string:lease)";
        finalize_request(*impl, nullptr, false);
        return TransportRequest{impl};
    }

    impl->req = req;
    return TransportRequest{impl};
}

TransportRequest
isend_serialized(Context & ctx,
                 int dest,
                 const std::vector<std::byte> & payload,
                 const TagLease & lease,
                 SendMode mode) noexcept
{
    auto impl = std::make_shared<TransportRequestImpl>();
    const MessageTag tag = lease.tag();
    impl->has_tag = true;
    impl->tag = tag;
    impl->use_ns = false;
    impl->alloc = nullptr;
    impl->pending_comm = ctx.comm();
    impl->is_fixed = false;
    impl->status_source = -1;
    impl->status_tag = static_cast<int>(tag);

    impl->buffer = std::make_shared<std::vector<std::byte>>(payload);

    MPI_Request req = MPI_REQUEST_NULL;
    const int rc = (mode == SendMode::Synchronous)
                       ? MPI_Issend(impl->buffer->data(),
                                    static_cast<int>(impl->buffer->size()),
                                    MPI_BYTE,
                                    dest,
                                    static_cast<int>(tag),
                                    ctx.comm(),
                                    &req)
                       : MPI_Isend(impl->buffer->data(),
                                   static_cast<int>(impl->buffer->size()),
                                   MPI_BYTE,
                                   dest,
                                   static_cast<int>(tag),
                                   ctx.comm(),
                                   &req);

    if (!detail::call_mpi(rc, "MPI_Isend/Issend failed (isend_serialized:bytes:lease)", ctx.comm()))
    {
        impl->error_msg = "MPI_Isend/Issend failed (isend_serialized:bytes:lease)";
        finalize_request(*impl, nullptr, false);
        return TransportRequest{impl};
    }

    impl->req = req;
    return TransportRequest{impl};
}
bool
TransportRequest::release() noexcept
{
    if (!_impl)
        return false;
    bool did = false;
    if (_impl->buffer)
    {
        _impl->buffer.reset();
        did = true;
    }
    if (_impl->payload_ptr)
    {
        _impl->payload_ptr = nullptr;
        did = true;
    }
    if (_impl->unpack_fn)
    {
        _impl->unpack_fn = nullptr;
        did = true;
    }
    if (_impl->has_tag && _impl->tag.has_value())
    {
        if (_impl->use_ns && _impl->tag_ns)
            _impl->tag_ns->release(*_impl->tag);
        else if (_impl->alloc)
            _impl->alloc->release(*_impl->tag);
        _impl->has_tag = false;
        _impl->tag.reset();
        did = true;
    }
    return did;
}
} // namespace compi